package com.max.spark.mysql

import com.max.spark.utils.{ConfUtil, PropsUtil, SparkUtil}
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql._

object SparkSQLMySQLApp {
  val confPath = "conf/mysql.properties"
  val hostname = ConfUtil.getProperty(confPath, "mysql.hostname")
  val port = ConfUtil.getProperty(confPath, "mysql.port")
  val username = ConfUtil.getProperty(confPath, "mysql.username")
  val password = ConfUtil.getProperty(confPath, "mysql.password")
  val driver = ConfUtil.getProperty(confPath, "mysql.driver")
  val db = ConfUtil.getProperty(confPath, "mysql.db")
  val table = ConfUtil.getProperty(confPath, "mysql.table")

  // 添加属性
  val properties = PropsUtil.getProps(driver, username, password)
  val url = s"jdbc:mysql://$hostname:$port/$db"
  // 添加属性第二种
  val prop = PropsUtil.getPropsByMap("mysql", driver, hostname, port, username, password, db, table)

  def main(args: Array[String]): Unit = {
    val spark = SparkUtil.sparkSession()
    import spark.implicits._
//    readType01(spark)
    readType02(spark)
//    readType03(spark)
//    write2mysql(spark)
    spark.stop()
  }

  // 读数据：第一种写法
  def readType01(spark: SparkSession): Unit = {
    val user = spark.read.jdbc(url, table, properties)
    user.show()
    spark.read.jdbc(url, s"(select max(age) from $table group by name) T", properties).show()
  }

  // 读数据：第二种写法
  def readType02(spark: SparkSession): Unit = {
    val df = spark.read.format("jdbc").options(prop).load()
    //读出来之后注册为临时表
    df.createOrReplaceTempView("student")
    df.show()
    //注册好之后就可以通过sql语句查询了
    spark.sql("select * from student where age>20").show()
  }

  // 读数据第三种
  def readType03(spark: SparkSession): Unit = {
    val reader: DataFrameReader = spark.read.format("jdbc")
      .option("url", url)
      .option("driver", driver)
      .option("user", username)
      .option("password", password)
      .option("dbtable", table)

    val source2: DataFrame = reader.load()
    source2.show()
  }

  // 写入数据
  def write2mysql(spark: SparkSession): Unit ={
    val user = spark.read.jdbc(url, "user", properties)
    val myuser = spark.read.format("jdbc").options(prop).load()
    user.createOrReplaceTempView("u")
    myuser.createOrReplaceTempView("v")
    val result = spark.sql("select u.address, u.sign_date, v.name, v.age from u,v  where u.username=v.name")
    result.show()
    // 第一种：将查询出的结果保存到mysql表之中
    result.write.mode(SaveMode.Append).jdbc(s"jdbc:mysql://$hostname:$port/$db","result", properties)

    // 第二种：将查询出的结果保存到mysql表之中
//    result.write.format("jdbc").options(prop).mode(SaveMode.Append).save()
  }




}
